package GradeTopN;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;
import java.util.Iterator;
import java.util.TreeMap;

public class GradeTopNMapper extends Mapper<LongWritable, Text, GradeTopNFlowBean, Text> {

    private TreeMap<GradeTopNFlowBean, Text> flowMap = new TreeMap<>();

    @Override
    protected void map(LongWritable key, Text value, Mapper.Context context) throws IOException, InterruptedException {
        // 1. 获取一行
        String line = value.toString();

        // 2. 切割
        String[] words = line.split(",");

        // 封装数据
        // 3. 封装数据
        String id = words[0];
//        int loadAmnt = Integer.parseInt(fields[1]);
//        int term = Integer.parseInt(fields[2]);
//        float interestRate = Float.parseFloat(fields[3]);
//        float installment = Float.parseFloat(fields[4]);
        String grade = words[5];
//        String subGrade = fields[6];
//        int employmentTitle = Integer.parseInt(fields[7]);
//        String employmentLength = fields[8];
//        int homeOwnership = Integer.parseInt(fields[9]);
        double annualIncome = Double.parseDouble(words[10]);
//        int verificationStatus = Integer.parseInt(fields[11]);
//        String issueDate = fields[12];
//        int purpose = Integer.parseInt(fields[13]);
//        int postCode = Integer.parseInt(fields[14]);
//        int regionCode = Integer.parseInt(fields[15]);
//        float dti = Float.parseFloat(fields[16]);
//        int delinquency_2years = Integer.parseInt(fields[17]);
//        int ficoRangeLow = Integer.parseInt(fields[18]);
//        int ficoRangeHigh = Integer.parseInt(fields[19]);
//        int openAcc = Integer.parseInt(fields[20]);
//        int pubRec = Integer.parseInt(fields[21]);
//        int pubRecBankruptcies = Integer.parseInt(fields[22]);
//        int revolBal = Integer.parseInt(fields[23]);
//        float revolUtil = Float.parseFloat(fields[24]);
//        int totalAcc = Integer.parseInt(fields[25]);
//        int initialListStatus = Integer.parseInt(fields[26]);
//        int applicationType = Integer.parseInt(fields[27]);
//        String earliesCreditLine = fields[28];
//        int title = Integer.parseInt(fields[29]);
//        int policyCode = Integer.parseInt(fields[30]);

//        Text keyOut = new Text(id);
//        GradeTopNFlowBean flowBean = new GradeTopNFlowBean(annualIncome, grade);
//        flowMap.put(flowBean, keyOut);

        Text valueOut = new Text(id);
        GradeTopNFlowBean flowBean = new GradeTopNFlowBean(annualIncome, grade);
        flowMap.put(flowBean, valueOut);

    }

    @Override
    protected void cleanup(Mapper<LongWritable, Text, GradeTopNFlowBean, Text>.Context context) throws IOException, InterruptedException {
        // 6 遍历treeMap集合，输出数据
        Iterator<GradeTopNFlowBean> iterator = flowMap.keySet().iterator();

        while (iterator.hasNext()) {
            GradeTopNFlowBean keyOut = iterator.next();
            Text valueOut = flowMap.get(keyOut);

            context.write(keyOut, valueOut);
        }
    }
}
